package wang.moshu.message;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.TaskRejectedException;

import javax.annotation.PostConstruct;

/**
 * 消息监听器
 *
 * @author dingxiangyong 2016年7月28日 下午2:12:08
 */
public abstract class AbstarctMessageHandler<T> implements Runnable {

    private static Log logger = LogFactory.getLog(AbstarctMessageHandler.class);

    @Autowired
    protected MessageTrunk messageTrunk;

    /**
     * 初始化启动本监听器
     */
    @PostConstruct
    public void startListen() {
        // 启动监听
        new Thread(this).start();
    }

    @Autowired
    private RedisUtil redisUtil;

    /**
     * 监听的消息类型
     */
    private String messageType;

    private boolean monitor;

    public AbstarctMessageHandler(String messageType) {
        this.messageType = messageType;
    }

    public void run() {
        while (true) {
            listen();
        }
    }

    /**
     * 监听消息
     */
    @SuppressWarnings("unchecked")
    public void listen() {
        // 阻塞获取redis列表值
        final Object obj = redisUtil.blpop(messageType, Integer.MAX_VALUE);

        // 如果获取失败，结束
        if (obj == null) {
            logger.warn("消息分发完毕");
            monitor = false;
            return;
        }

        if (!monitor) {
            logger.warn("消息分发开始");
            monitor = true;
        }

        try {
            messageTrunk.getThreadPool().submit(new Runnable() {

                public void run() {
                    Message<T> message = (Message<T>) obj;
                    try {
                        // 如果获取成功，则交给子类做业务处理
                        handle(message.getContent());
                    } catch (Exception ex) {
                        logger.error(ex);
                        // 处理失败，判断是否需要重试
                        if (message.getFailTimes().intValue() >= messageTrunk.getFailRetryTimes()) {
                            handleFailed(message.getContent());
                        } else {
                            message.setFailTimes(message.getFailTimes().intValue() + 1);
                            // 再次put回消息总线，等待下次重试
                            messageTrunk.put(message);

                            if (logger.isDebugEnabled()) {
                                StringBuilder sb = new StringBuilder();
                                sb.append("msg:[").append(message).append("], 执行失败，准备重试。");
                                logger.debug(sb.toString());
                            }
                        }
                    }
                }

            });

        } catch (TaskRejectedException ex) {
            logger.warn("线程池已满，准备回写任务，暂停本线程");
            // 如果发生任务拒绝加入线程池，则回写任务到redis，等待重试
            messageTrunk.put((Message<T>) obj);

            // 暂停指定时间
            try {
                Thread.sleep(messageTrunk.getThreadPoolFullSleepSeconds() * 1000);
            } catch (InterruptedException e) {
                logger.warn("生产者暂停异常", ex);
            }
        } catch (Exception ex) {
            logger.error("消息总线发生异常", ex);
        }
    }

    /**
     * 获取到消息后做业务处理
     */
    public abstract void handle(T obj);

    /**
     * 消息多次重试处理失败
     */
    public abstract void handleFailed(T obj);
}
